package com.thiswhyme.republish.demo.rabbitmq;

import com.rabbitmq.client.Channel;
import com.thiswhyme.republish.rabbitmq.core.AbstractRetriableReceiver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import java.io.IOException;

/**
 * @description: 可重试消费者demo
 * @author: thiswhyme
 * @date: 2021/7/28
 */
@Slf4j
public class DemoReceiver extends AbstractRetriableReceiver {

    @StreamListener(DemoChannel.INPUT_CHANNEL_RETRY)
    @Override
    public void onMessage(Message<String> message) {
        Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        if (!"【Retry测试-重试】测试消息".equals(message.getPayload())) {
            log.info("监听到INPUT_CHANNEL_RETRY消息，内容：{}", message.getPayload());
        } else {
            log.info("监听到INPUT_CHANNEL_RETRY消息，内容：{}", message.getPayload());
            retry(message, DemoChannel.INPUT_CHANNEL_RETRY, "我是自定义参数");
        }
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.warn("INPUT_CHANNEL_RETRY消息，消息ACK失败，内容：{}", message.getPayload());
        }
    }

    @Override
    public void doIfRetryFail(Object bizParams) {
        log.info("重试2次仍然失败，执行自定义逻辑，参数：{}", bizParams);
    }
}
